package com.bw.app.functions;

import com.alibaba.fastjson.JSONObject;
import com.bw.bean.OrderWide;
import com.bw.common.GmallConfig;
import com.bw.utils.DimUtil;
import com.bw.utils.ThreadPoolUtil;
import lombok.SneakyThrows;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.lang.reflect.InvocationTargetException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.text.ParseException;
import java.util.Collections;

public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimAsyncInterface<T> {

    private Connection connection;
    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        // 做数据库连接
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    @Override
    public void asyncInvoke(T t, ResultFuture<T> resultFuture) throws Exception {
        // 线程池  单例模式--双重校验
        ThreadPoolUtil.getThreadPoolExecutor().submit(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {
                String id = getKey(t);
//                System.out.println("id = " + id);
                // 去查询维度
                JSONObject dimInfo = DimUtil.getDimInfo(connection, tableName, id);
//                System.out.println("dimInfo = " + dimInfo);
                if (dimInfo != null) {
                    join(t,dimInfo);
                }
                // 返回结果
                resultFuture.complete(Collections.singletonList(t));
            }
        });


    }


    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println("超时了....");
    }
}
